package com.suxin.module.sender.consumer;

import com.suxin.module.MyOpenTSDBConfig;
import com.suxin.module.bean.MyPoint;
import com.suxin.module.http.MyHttpClient;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

/**
 * @Classname MyConsumerImpl
 * @Description [ TODO ]
 * @Author Tang
 * @Date 2020/10/16 10:48
 * @Created by ASUS
 */
@Slf4j
public class MyConsumerImpl implements MyConsumer{

    private BlockingQueue<MyPoint> blockingQueue;

    private MyHttpClient myHttpClient;

    private MyOpenTSDBConfig myOpenTSDBConfig;

    private ThreadPoolExecutor threadPoolExecutor;

    private CountDownLatch countDownLatch;

    public MyConsumerImpl(BlockingQueue<MyPoint> blockingQueue,
                          MyHttpClient myHttpClient,
                          MyOpenTSDBConfig myOpenTSDBConfig) {
        this.blockingQueue = blockingQueue;
        this.myHttpClient = myHttpClient;
        this.myOpenTSDBConfig = myOpenTSDBConfig;
        // 获取消费者线程数量
        int threadCount = myOpenTSDBConfig.getPutConsumerThreadCount();
        int[] i = new int[1];
        // 显示创建线程池
        this.threadPoolExecutor = new ThreadPoolExecutor(
                threadCount,
                threadCount,
                1000L,
                TimeUnit.MICROSECONDS,
                new LinkedBlockingDeque<>(threadCount),
                (r) -> {
                    return new Thread(r,"myThreadPoolExecutor...." + i[0]++);
                }
        );
        this.countDownLatch = new CountDownLatch(threadCount);
    }

    @Override
    public void start() {
        for (int i = 0; i < myOpenTSDBConfig.getPutConsumerThreadCount(); i++) {
            // 提交任务到线程池
            this.threadPoolExecutor.execute(
                    new MyConsumerRunnable(
                            blockingQueue,
                            myHttpClient,
                            myOpenTSDBConfig,
                            countDownLatch
                    )
            );
        }
    }

    @Override
    public void gracefulStop() {
        this.forceStop(false);
    }

    @Override
    public void forceStop(boolean isForce) {
        if (!isForce) {
            // 强制退出不等待，截断消费者线程
            this.threadPoolExecutor.shutdown();
        }else {
            // 截断消费者线程
            while (!threadPoolExecutor.isShutdown() ||
                    // 调用shutdown之后,但是线程池还未完全终止返回true
                    !threadPoolExecutor.isTerminated()) {
                this.threadPoolExecutor.shutdown();
            }
            // 等待所有消费者线程结束
            try {
                this.countDownLatch.await();
            } catch (InterruptedException e) {
                log.error("An error occurred waiting for the consumer thread to close", e);
            }
        }
    }
}